#include "unp.h"


static void sig_usr1(int signo);

mqd_t mqd;
void  *buff;
struct mq_attr attr;
struct sigevent sigev;
int fd[2] = {0};

int main(int argc,char *argv[])
{
   fd_set rset,eset;
   sigset_t sigset;
   sigemptyset(&sigset);
   sigaddset(&sigset,SIGUSR1);

   mqd = mq_open(argv[1],O_RDONLY | O_NONBLOCK);
   if(mqd == -1)
     err_sys("mq_open error");
   if(mq_getattr(mqd,&attr) == -1)
     err_sys("mq_getattr error");
 
   buff = malloc(attr.mq_msgsize);

   signal(SIGUSR1,sig_usr1);
   sigev.sigev_notify=SIGEV_SIGNAL;
   sigev.sigev_signo =SIGUSR1;
   if(mq_notify(mqd,&sigev) ==-1)
     err_sys("mq_notify error");

   if(pipe(fd) == -1)
     err_sys("pipe error");
   
   FD_ZERO(&rset);
   FD_ZERO(&eset);
   for(;;){
     FD_SET(fd[0],&rset); 
     FD_SET(fd[0],&eset); 
     printf("select begin\n");
     //int rc = pselect(fd[0]+1,&rset,NULL,NULL,NULL,&sigset);
     int rc = select(fd[0]+1,&rset,NULL,&eset,NULL);
     printf("select end\n");
     if( rc == -1 ){
       err_msg("select error %d", FD_ISSET(fd[0],&eset));
       if(FD_ISSET(fd[0],&eset)){
          goto mq_re;
       }
     }else if(rc == 0){
       err_msg("select timeout!");
     }
     mq_re:
     if(FD_ISSET(fd[0],&rset) || FD_ISSET(fd[0],&eset)){
       printf("sig_usr1 begin\n");
       read(fd[0],buff,1);
       ssize_t n;
       u_int32_t prio;
       if(mq_notify(mqd,&sigev) ==-1)
         err_sys("mq_notify error");
       while((n=mq_receive(mqd,buff,attr.mq_msgsize,&prio)) !=-1){
          printf("SIGUSR1 received, read %ld bytes\n",(long)n);
       }
       if(errno != EAGAIN)
         err_sys("mq_receive error");
     }
   }
   exit(0);
}

static void sig_usr1(int signo){
   printf("SIGUSR1 received\n");
   if(write(fd[1]," ",1)!=1)
     err_sys("write error");
   return;
}
